package com.czm.art_light.manager.task;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.Time;
import co.elastic.clients.elasticsearch.core.BulkRequest;
import co.elastic.clients.elasticsearch.core.BulkResponse;
import co.elastic.clients.elasticsearch.core.bulk.BulkResponseItem;
import com.czm.art_light.constant.CommonConsts;
import com.czm.art_light.constant.EsConsts;
import com.czm.art_light.mapper.ArtArtistMapper;
import com.czm.art_light.mapper.ArtWorkMapper;
import com.czm.art_light.param.response.ArtArtistEsRespDto;
import com.czm.art_light.param.response.ArtWorkEsRespDto;
import com.czm.art_light.service.ArtArtistService;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.handler.annotation.XxlJob;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.time.LocalDateTime;
import java.util.List;

/**
 * 艺术家数据同步到 elasticsearch 任务
 */
@Component
@RequiredArgsConstructor
@Slf4j
public class ArtArtistToEsTask {
    @Autowired
    private ArtArtistService artArtistService;

    @Autowired
    private ElasticsearchClient elasticsearchClient;

    // 记录上次同步的时间戳
    private LocalDateTime lastSyncTime = LocalDateTime.now();

    /**
     * 每月凌晨做一次全量数据同步
     * 每天凌晨做一次增量数据同步
     */
    @SneakyThrows
    @XxlJob("saveArtArtistToEsJobHandler")
    public ReturnT<String> saveToEs() {
        log.info("艺术家数据同步到 elasticsearch 任务开始");
        try {
            // 判断是否需要进行全量同步
            if (isFullSyncRequired()) {
                // 全量同步
                fullSync();
            } else {
                // 增量同步
                incrementalSync();
            }
            log.info("艺术家数据同步到 elasticsearch 任务结束");
            return ReturnT.SUCCESS;
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            return ReturnT.FAIL;
        }
    }

    /**
     * 判断是否需要进行全量同步
     */
    private boolean isFullSyncRequired() {
        // 假设每月第一天凌晨进行全量同步
        LocalDateTime now = LocalDateTime.now();
        return now.getDayOfMonth() == 1 && now.getHour() == 0 && now.getMinute() == 0;
    }

    /**
     * 全量数据同步
     */
    private void fullSync() throws IOException {
        log.info("开始全量同步艺术家数据");
        // 查询当前所有的艺术家数据
        List<ArtArtistEsRespDto> artArtistEsRespDtos = artArtistService.syncArtistsToES();

        // 批量导入 Elasticsearch
        bulkIndexData(artArtistEsRespDtos);

        // 更新上次同步时间
        lastSyncTime = LocalDateTime.now();
        log.info("全量同步艺术家数据完成");
    }

    /**
     * 增量数据同步
     */
    private void incrementalSync() throws IOException {
        log.info("开始增量同步艺术家数据");
        // 查询自上次同步以来新增或修改的艺术家数据
        List<ArtArtistEsRespDto> modifiedArtists = artArtistService.getModifiedArtists(lastSyncTime);

        // 批量导入 Elasticsearch
        bulkIndexData(modifiedArtists);

        // 更新上次同步时间
        lastSyncTime = LocalDateTime.now();
        log.info("增量同步艺术家数据完成");
    }

    /**
     * 批量导入数据到 Elasticsearch
     */
    private void bulkIndexData(List<ArtArtistEsRespDto> data) throws IOException {
        if (data.isEmpty()) {
            log.info("没有需要同步的数据");
            return;
        }

        BulkRequest.Builder br = new BulkRequest.Builder();
        for (ArtArtistEsRespDto artArtistEsRespDto : data) {
            br.operations(op -> op
                    .index(idx -> idx
                            .index(EsConsts.ArtistIndex.INDEX_NAME)
                            .id(artArtistEsRespDto.getId().toString())
                            .document(artArtistEsRespDto))
            ).timeout(Time.of(t -> t.time(CommonConsts.ELASTICSEARCH_ADD_TIMEOUT)));
        }

        BulkResponse result = elasticsearchClient.bulk(br.build());

        // 记录错误日志
        if (result.errors()) {
            log.error("Bulk had errors");
            for (BulkResponseItem item : result.items()) {
                if (item.error() != null) {
                    log.error(item.error().reason());
                }
            }
        }
    }
}